package org.exhaust.framework.net.client.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.exhaust.framework.net.common.protocol.ChannelEventDispatcher;
import org.exhaust.framework.net.common.protocol.EventLoopFactory;
import org.exhaust.framework.net.common.protocol.RunningState;
import org.exhaust.framework.net.common.protocol.decoder.tcp.NoneSharableTcpByteBufDecoder;
import org.exhaust.framework.net.common.protocol.encoder.TcpFrameToByteBufEncoder;
import org.exhaust.framework.net.common.protocol.handler.ConnectionMonitorHandler;
import org.exhaust.framework.net.common.protocol.handler.idle.ChannelReaderIdleEventHandler;
import org.exhaust.framework.net.common.protocol.handler.idle.ChannelReaderIdleStateHandler;
import org.exhaust.framework.net.common.protocol.handler.idle.ChannelWriterIdleEventHandler;
import org.exhaust.framework.net.common.protocol.handler.idle.ChannelWriterIdleStateHandler;

import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

@Slf4j
public class BaseTcpClient implements TcpClient {

    private volatile Channel channel = null;

    private volatile RunningState state = RunningState.TERMINATED;

    private ChannelEventDispatcher eventDispatcher;

    private TcpClientProperty tcpClientProperty;

    private Bootstrap bootstrap;

    private EventLoopGroup eventLoopGroup;

    private volatile int retriedTimes = 0;

    private Supplier<NoneSharableTcpByteBufDecoder> decoderSupplier;

    private TcpFrameToByteBufEncoder encoder;

    private ConnectionMonitorHandler connectionMonitorHandler = null;

    /**
     * @param decoderSupplier   supply中必须每次都new一个新对象，因为ByteToMessageDecoder不是线程安全的
     * @param encoder
     * @param eventDispatcher
     * @param tcpClientProperty
     */
    public BaseTcpClient(Supplier<NoneSharableTcpByteBufDecoder> decoderSupplier, TcpFrameToByteBufEncoder encoder, ChannelEventDispatcher eventDispatcher, TcpClientProperty tcpClientProperty) {
        this.decoderSupplier = decoderSupplier;
        this.encoder = encoder;
        this.eventDispatcher = eventDispatcher;
        this.tcpClientProperty = tcpClientProperty;
        this.bootstrap = new Bootstrap();
        this.eventLoopGroup = EventLoopFactory.eventLoopGroup(1, "Client-Worker-EventLoopGroup");
        this.connectionMonitorHandler = new ConnectionMonitorHandler(eventDispatcher);
    }


    @Override
    public void connect() {
        this.releaseChannel();
        final ChannelFuture cf = bootstrap.connect(this.tcpClientProperty.getServerAddress(), this.tcpClientProperty.getServerPort());
        cf.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    //重连交给后端线程执行
                    future.channel().eventLoop().schedule(() -> {
                        retriedTimes++;
                        log.warn("连接失败，正在进行第{}尝试,服务器地址：{},服务器端口:{}", retriedTimes, tcpClientProperty.getServerAddress(), tcpClientProperty.getServerPort());
                        try {
                            connect();
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    }, 3000, TimeUnit.MILLISECONDS);
                } else {
                    retriedTimes = 0;
                    channel = future.channel();
                }
            }
        });
    }

    @Override
    public Channel getChannel() {
        return this.channel;
    }

    @Override
    public EventLoopGroup getEventLoopGroup() {
        return eventLoopGroup;
    }

    @Override
    public RunningState getState() {
        return this.state;
    }

    private void releaseChannel() {
        if (this.channel != null) {
            final ChannelPipeline pipeline = channel.pipeline();
            while (pipeline.first() != null) {
                pipeline.removeFirst();
            }
            try {
                this.channel.close();
            } catch (Throwable e) {
                log.warn("close channel failed:{}", e.getMessage());
            }
            this.channel = null;
        }
    }

    @Override
    public void start() throws Exception {
        this.state = RunningState.RUNNING;
        bootstrap.group(eventLoopGroup)
                .channel(EventLoopFactory.socketChannelClass())
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.tcpClientProperty.getConnectTimeout())
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        if (tcpClientProperty.getReaderIdleSeconds() > 0) {
                            pipeline.addLast("ClientReaderIdleStateHandler", new ChannelReaderIdleStateHandler(tcpClientProperty.getReaderIdleSeconds()));
                            pipeline.addLast("ClientReaderIdleEventHandler", new ChannelReaderIdleEventHandler(eventDispatcher));
                        }
                        if (tcpClientProperty.getWriterIdleSeconds() > 0) {
                            pipeline.addLast("ClientWriterIdleStateHandler", new ChannelWriterIdleStateHandler(tcpClientProperty.getWriterIdleSeconds()));
                            pipeline.addLast("ClientWriterIdleEventHandler", new ChannelWriterIdleEventHandler(eventDispatcher));
                        }
                        pipeline.addLast("ConnectionMonitorHandler", connectionMonitorHandler);
                        if (decoderSupplier != null) {
                            final NoneSharableTcpByteBufDecoder decoder = decoderSupplier.get();
                            if (decoder != null) {
                                pipeline.addLast("FrameDecoderAndHandler", decoder);
                            }
                        }
                        if (encoder != null) {
                            pipeline.addLast("FrameEncoder", encoder);
                        }

                    }
                });
        this.connect();
    }

    @Override
    public void stopGracefully() {
        this.state = RunningState.STOPPING;
        this.releaseChannel();
        this.eventLoopGroup.shutdownGracefully();
        this.eventLoopGroup = null;
        this.bootstrap = null;
        this.state = RunningState.TERMINATED;
    }
}
